package com.example.mqtt.config;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * 订阅
 * <pre>
 * DirectChannel
 * PublishSubscribeChannel
 * ExecutableChannel
 * PriorityChannel
 * RendezvousChannel：点对点，适用于需要确保消息被立即处理的场景
 * </pre>
 */
@Configuration
@Slf4j
public class MqttInboundConfig {
    @Autowired
    private ThreadPoolTaskExecutor pool;
    @Resource
    private MqttProperty mqttProperty;
    @Resource
    private MqttPahoClientFactory mqttPahoClientFactory;

    @Bean
    public MessageChannel mqttInboundChannel() {
        return new ExecutorChannel(pool);
    }


    // 将 topic 绑定到 通道
    @Bean
    public MessageProducer messageProducer() {
        MqttPahoMessageDrivenChannelAdapter p = new MqttPahoMessageDrivenChannelAdapter(
                mqttProperty.getUrl(),
                mqttProperty.getSubClientId(),
                mqttPahoClientFactory,
                mqttProperty.getDefaultTopic());

        p.setQos(mqttProperty.getQos());
        p.setConverter(new DefaultPahoMessageConverter());
        p.setOutputChannel(mqttInboundChannel());
        return p;
    }

}
